#!/bin/bash
# File Name   : cg_monitor.sh
# Author      : moshan
# Mail        : mo_shan@yeah.net
# Version     : 1.0
# Created Time: 2025-03-12 11:08:00
# Function    : cgroup监控采集
#########################################################################
set -e -o pipefail

work_dir="/opt/soft/cgroup"
conf_dir="${work_dir}/conf"

localhost_ip="$(/sbin/ip a 2>/dev/null|grep inet|grep brd|grep scope|awk '{print $2}'|awk -F'/' '{print $1}'|head -1)" || true

export PATH="/opt/soft/mysql80/bin:/opt/soft/mysql57/bin:${PATH}"

. /etc/profile || true

env="${1}"

global_master_mark="false"
global_tidb_mark="false"

run_date="$(date +%F)" #每天重启一次

function f_write_monitor_info_to_tidb()
{
    declare sql
    sql="${1}"
    for ((i=0;i<10;i++))
    do
        echo ${sql} | timeout 10s ${tidb_comm} && break
        sleep 1
    done

}

function f_get_disk_num()
{
    declare mp

    mp="${1}"

    lsblk -o NAME,MAJ:MIN,RM,SIZE,RO,TYPE,MOUNTPOINT | awk -v mp="${mp}" -F "[[:space:]]+" '
    
    NR == 1 { next; }
    
    $6 == "disk" {
    
        disk = $1;
    
        num[disk] = $2;
    
        if ($NF) {
    
            gsub(/^[[:space:]]+|[[:space:]]+$/, "", $NF);
    
            if($NF==mp) {print(num[disk])};
    
        }
    }
    
    $6 == "part" && $NF {
    
        gsub(/^[[:space:]]+|[[:space:]]+$/, "", $NF);
    
        if($NF==mp) {print(num[disk])};
    
    }'
}

function f_monitor_cpu()
{
    declare opt db port isovalue pid fpath cpath upath usagemem susagemem
    opt="${1}"
    db="${2}"
    port="${3}"
    isovalue="$((${4} * 100))"
    pid="${5}"

    fpath="/sys/fs/cgroup/${opt}/${db}${port}/cpuacct.usage"
    cpath="/sys/fs/cgroup/${opt}/${db}${port}/cgroup.procs"
    upath="/sys/fs/cgroup/memory/${db}${port}/memory.usage_in_bytes"

    [ ! -f "${cpath}" ] && return 1
    grep -w ${pid} ${cpath} || cgclassify -g ${opt}:${db}${port} ${pid}

    [ ! -f ${fpath} ] && return 1

    [ "${opt}x" != "cpux" ] && return 0

    now_time="$(date '+%F %T')"
    stime="$(date +%s%N)"
    scpu="$(cat ${fpath})"

    if [ -f "${upath}" ]
    then
        susagemem="$(cat ${upath})"
    else
        susagemem=0
    fi

    sleep 1

    ecpu="$(cat ${fpath})"
    etime="$(date +%s%N)"

    if [ -f "${upath}" ]
    then
        usagemem="$(awk -v usa="${susagemem}" '{if ($1>usa) print int(($1-usa)/1024/1024);else print 0}' ${upath})"
    else
        usagemem=0
    fi

    if [ -z "${stime}" ] || [ -z "${etime}" ] || [ -z "${scpu}" ] || [ -z "${ecpu}" ]
    then
        return 1
    fi

    cpu="$(echo "${stime} ${etime} ${scpu} ${ecpu}"|awk '{print int((($4-$3)/($2-$1))*100)}')"

    #echo "insert into tb_monitor_cgroup_info(rshost,rsport,isotype,mvalue,isovalue,a_time) value('${localhost_ip}',${port},'cpu',${cpu},${isovalue} * 100,'${now_time}');" |${tidb_comm}
    f_write_monitor_info_to_tidb "replace into tb_monitor_cgroup_info(rshost,rsport,isotype,mvalue,isovalue,a_time) value('${localhost_ip}',${port},'cpu',${cpu},${isovalue},'${now_time}');
        replace into tb_monitor_cgroup_info(rshost,rsport,isotype,mvalue,isovalue,a_time) value('${localhost_ip}',${port},'usagemem',${usagemem}, 100,'${now_time}');
        replace into tb_monitor_cgroup_info(rshost,rsport,isotype,mvalue,isovalue,a_time) value('${localhost_ip}',${port},'susagemem',${susagemem}, 100,'${now_time}');"
}

function f_monitor_load()
{
    declare load isovalue
    now_time="$(date '+%F %T')"
    load="$(uptime |awk -F 'load average:' '{print $2}'|awk -F',' '{print int($1)}')"
    isovalue="$(lscpu |grep "^CPU(s):"|awk '{print int($2)}')"
    f_write_monitor_info_to_tidb "replace into tb_monitor_cgroup_info(rshost,rsport,isotype,mvalue,isovalue,a_time) value('${localhost_ip}',0,'load',${load},${isovalue},'${now_time}');"
}

function f_monitor_cpu_memory()
{
    declare db port pid mvalue now_time cpu memory opt cpath isovalue
    db="${1}"
    port="${2}"
    pid="${3}"
    opt="${4}"
    isovalue="$((${5}*1024))"
    cpath="/sys/fs/cgroup/${opt}/${db}${port}/cgroup.procs"

    [ ! -f "${cpath}" ] && return 1

    grep -w ${pid} ${cpath} || cgclassify -g ${opt}:${db}${port} ${pid}

    now_time="$(date '+%F %T')"

    #mvalue="$(top -p ${pid} -b -n 1|grep -w ${pid}|tail -1|awk '{print $9,$10}')"
    mvalue="$(top -p ${pid} -b n 1|grep -Pw "(^%Cpu\(s\):|^[ ]{0,}${pid})"|awk '{print $9,$10}'|tail -2|tr "\n" " ")"

    iowait="$(awk '{print int($2)}' <<< "${mvalue}")"

    memory="$(free|grep -w Mem:|awk '{print $2}')"

    memory="$(awk -v m=${memory} '{print int($4*m/1024/100)}' <<< "${mvalue}")"

    f_write_monitor_info_to_tidb "replace into tb_monitor_cgroup_info(rshost,rsport,isotype,mvalue,isovalue,a_time) value('${localhost_ip}',1,'iowait',${iowait},50,'${now_time}');
        replace into tb_monitor_cgroup_info(rshost,rsport,isotype,mvalue,isovalue,a_time) value('${localhost_ip}',${port},'memory',${memory},${isovalue},'${now_time}');"
}

function f_monitor_iops()
{
    declare db port fpath mvalue now_time cpath isovalue
    db="${1}"
    port="${2}"
    opt="${3}"
    fpath="${4}"
    pid="${5}"
    isovalue="${6}"
    diskid="${7}"

    fpath="/sys/fs/cgroup/blkio/${db}${port}/${fpath}"

    cpath="/sys/fs/cgroup/blkio/${db}${port}/cgroup.procs"

    [ ! -f "${cpath}" ] && return 1

    grep -w ${pid} ${cpath} || cgclassify -g blkio:${db}${port} ${pid}

    [ ! -f "${fpath}" ] && return 1

    now_time="$(date '+%F %T')"

    his="$(grep -P "^${diskid} (Read|Write)" ${fpath}|tr "\n" " "|awk '{print $3,$6}')"

    sleep 1

    now="$(grep -P "^${diskid} (Read|Write)" ${fpath}|tr "\n" " "|awk '{print $3,$6}')"

    his_read="$(awk '{print $1}' <<< "${his}")"
    his_write="$(awk '{print $2}' <<< "${his}")"


    if [ "${opt}x" == "iox" ]
    then
        now_read="$(awk -v h="${his_read}" '{print int(($1-h)/1024/1024)}' <<< "${now}")"
        now_write="$(awk -v h="${his_write}" '{print int(($2-h)/1024/1024)}' <<< "${now}")"
    else
        now_read="$(awk -v h="${his_read}" '{print $1-h}' <<< "${now}")"
        now_write="$(awk -v h="${his_write}" '{print $2-h}' <<< "${now}")"
    fi

    f_write_monitor_info_to_tidb "replace into tb_monitor_cgroup_info(rshost,rsport,isotype,mvalue,isovalue,a_time) value('${localhost_ip}',${port},'${opt}_read',${now_read},${isovalue},'${now_time}');
          replace into tb_monitor_cgroup_info(rshost,rsport,isotype,mvalue,isovalue,a_time) value('${localhost_ip}',${port},'${opt}_write',${now_write},${isovalue},'${now_time}');"
}

function f_monitor_net()
{
    return 0
}

function f_check_tidb()
{
    declare exe_state is_master
    exe_state="0"
    is_master="$(timeout 5s ${tidb_comm} -NBe "select 'tidb'" 2>/dev/null|grep -w 'tidb'|tail -1)" || exe_state="${?}"

    if [ "${exe_state}x" != "0x" ] 
    then
        exe_state="0"
        is_master="$(timeout 5s ${local_tidb_comm} -NBe "select 'tidb'" 2>/dev/null|grep -w 'tidb'|tail -1)" || exe_state="${?}"
        tidb_comm="${local_tidb_comm}"
        global_tidb_mark="true" #本机是4097的主库，不能连clb访问
    fi

    if [ "${is_master}x" != "tidbx" ] || [ "${exe_state}x" != "0x" ]
    then
        return 1
    fi

    return 0
}

function f_check_master()
{
    declare exe_state is_master
    exe_state="0"
    is_master="$(timeout 5s ${mysql_comm} -NBe "show slave status\G" 2>/dev/null|wc -l)" || exe_state="${?}"

    if [ "${exe_state}x" != "0x" ] 
    then
        exe_state="0"
        is_master="$(timeout 5s ${local_mysql_comm} -NBe "show slave status\G" 2>/dev/null|wc -l)" || exe_state="${?}"
        mysql_comm="${local_mysql_comm}"
        global_master_mark="true" #本机是4097的主库，不能连clb访问
    fi

    if [ "${is_master}x" != "0x" ] || [ "${exe_state}x" != "0x" ]
    then
        return 1
    fi

    return 0
}

function f_insert_monitor_to_his()
{

    declare sql
    sql="delete from tb_monitor_cgroup_info_his where a_time <= date_add(now(), INTERVAL - 6 MINUTE) and rshost = '${localhost_ip}';
        insert into tb_monitor_cgroup_info_his(rshost,rsport,isotype,mvalue,isovalue,a_time) select rshost,rsport,isotype,mvalue,isovalue,a_time from tb_monitor_cgroup_info where rshost = '${localhost_ip}' and a_time >= date_add(now(), INTERVAL - 5 MINUTE)"
    echo ${sql} | timeout 60s ${tidb_comm}
}

function f_main()
{
    if [ "${env}x" != "devx" ] && [ "${env}x" != "oltpx" ]
    then
        return 1
    else
        source ${conf_dir}/${env}.conf
    fi

    if [ "${global_master_mark}x" == "falsex" ]
    then
        mysql_comm="${mysql_path} -u${mysql_user} -p${mysql_pass} -h${mysql_host} -P ${mysql_port} ${mysql_db}"
        tidb_comm="${mysql_path} -u${mysql_user} -p${mysql_pass} -h${tidb_host} -P ${tidb_port} ${tidb_db}"
    fi

    local_mysql_comm="${mysql_path} -u${mysql_user} -p${mysql_pass} -h${localhost_ip} -P ${mysql_port} ${mysql_db}"
    local_tidb_comm="${mysql_path} -u${mysql_user} -p${mysql_pass} -h${tidb_host} -P ${tidb_port} ${tidb_db}"

    f_check_master

    f_check_tidb

    declare -a info_list

    sql="SELECT concat(rsport,'@@@',isotype,'@@@',isovalue,'@@@',istate,'@@@',pidpath,'@@@',mountpoin,'@@@',dbtype) 
         from tb_cgroup_manager 
         where rshost = '${localhost_ip}' and istate = 1 and rule = 'add';"

    info_list=($(echo "${sql}" |timeout 60s ${mysql_comm} -NB 2>/dev/null))

    # 允许的进程数
    THREAD_NUM=10
    PPATH="/opt/soft/cgroup"
    pipe_file="${PPATH}/pipe.mark" 
    [ -p "${pipe_file}" ] && rm -f "${pipe_file}"
    mkfifo ${pipe_file}
    # 定义描述符为3的管道
    exec 3<> ${pipe_file}
    
    # 预先写入指定数量的进程个数
    seq ${THREAD_NUM} 1>&3

    for info in ${info_list[@]}
    do
        read -u 3
        {
            rsport="$(awk -F'@@@' '{print $1}' <<< "${info}")"
            isotype="$(awk -F'@@@' '{print $2}' <<< "${info}")"
            isovalue="$(awk -F'@@@' '{print $3}' <<< "${info}")"
            istate="$(awk -F'@@@' '{print $4}' <<< "${info}")"
            pidpath="$(awk -F'@@@' '{print $5}' <<< "${info}")"

            if [ -f "${pidpath}" ]
            then
                pid="$(cat ${pidpath})"
            else
                which lsof0 >/dev/null 2>&1 || yum -y install lsof > /dev/null 2>&1
                pid="$(lsof -i:3574 -t)"
            fi

            mountpoin="$(awk -F'@@@' '{print $6}' <<< "${info}")"
            dbtype="$(awk -F'@@@' '{print $7}' <<< "${info}")"

            #if [ "${isotype}x" == "mem_oomx" ] || [ "${isotype}x" == "memx" ]
            if [ "${isotype}x" == "mem_oomx" ] || [ "${isotype}x" == "memx" ] || [ "${isotype}x" == "memory_oomx" ] || [ "${isotype}x" == "memoryx" ]

            then
                #[ "${isotype}x" == "mem_oomx" -o "${isotype}x" == "memx" ] && isotype="memory"
                isotype="memory"
                f_monitor_cpu_memory "${dbtype}" "${rsport}" "${pid}" "${isotype}" "${isovalue}" || true
            elif [ "${isotype}x" == "cpux" ]
            then
                f_monitor_cpu "${isotype}" "${dbtype}" "${rsport}" "${isovalue}" "${pid}" || true
            elif [ "${isotype}x" == "io_readx" ] || [ "${isotype}x" == "io_writex" ]
            then
                f_monitor_iops "${dbtype}" "${rsport}" "io" "blkio.throttle.io_service_bytes" "${pid}" "${isovalue}" "$(f_get_disk_num "${mountpoin}")" || true
            elif [ "${isotype}x" == "iops_readx" ] || [ "${isotype}x" == "iops_writex" ]
            then
                f_monitor_iops "${dbtype}" "${rsport}" "iops" "blkio.throttle.io_serviced" "${pid}" "${isovalue}" "$(f_get_disk_num "${mountpoin}")" || true
            elif [ "${isotype}x" == "net_inx" ] || [ "${isotype}x" == "net_outx" ]
            then
                f_monitor_net "net" "${dbtype}" "${rsport}"
            else
                echo 1>&3
                continue
            fi
            echo 1>&3
        } &
    done
    wait
    [ -p "${pipe_file}" ] && rm -f "${pipe_file}"

    f_monitor_load || true

    f_insert_monitor_to_his || true
}

stop_file="/opt/soft/cgroup/stop_file"

while :;
do
    [ -f "/opt/soft/cgroup/stop_file_long" ] && break
    loop_start="$(date +%s)"
    f_main
    [ "${run_date}x" != "$(date +%F)x" ] && break
    [ -f "${stop_file}" ] && rm -f ${stop_file} && break
    sleep $((15-($(date +%s)-${loop_start}))) >/dev/null || true
done

